package com.yifeng.repo.controller.kafka.web;

import com.gomcarter.frameworks.base.context.SpringContextHolder;
import com.gomcarter.frameworks.base.json.ErrorCode;
import com.yifeng.repo.base.utils.servlet.RequestHelper;
import com.yifeng.repo.base.utils.servlet.ResponseHelper;
import com.yifeng.repo.controller.kafka.message.KafkaProduceManager;
import com.yifeng.repo.controller.kafka.message.KafkaPullConsumeManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.PartitionInfo;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * Created by daibing on 2024/5/2.
 */
@Slf4j
public class KafkaServlet extends HttpServlet {
    public static final String PRODUCE_MSG_MQ = "/produce/listPartitionInfo";
    public static final String PULL_CONSUME_MSG_MQ = "/pullConsume/listPartitionInfo";
    public static final String PULL_CONSUME_MSG_MQ_STATUS = "/pullConsume/listPartitionStats";
    private final String mqAdminPrefix;

    public KafkaServlet(String mqAdminPrefix) {
        this.mqAdminPrefix = mqAdminPrefix;
    }

    @Override
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        String uri = req.getRequestURI();
        int index = uri.indexOf(mqAdminPrefix + "/") + mqAdminPrefix.length() + 1;
        try {
            if (uri.endsWith(PRODUCE_MSG_MQ)) {
                String topic = uri.substring(index, uri.indexOf(PRODUCE_MSG_MQ));
                List<PartitionInfo> partitionInfos = SpringContextHolder.getBean(KafkaProduceManager.class).listPartitionInfo(topic);
                ResponseHelper.doWrite(resp, partitionInfos);
            } else if (uri.endsWith(PULL_CONSUME_MSG_MQ)) {
                String topic = uri.substring(index, uri.indexOf(PULL_CONSUME_MSG_MQ));
                List<PartitionInfo> partitionInfos = SpringContextHolder.getBean(KafkaPullConsumeManager.class).listPartitionInfo(topic);
                ResponseHelper.doWrite(resp, partitionInfos);
            }  else if (uri.endsWith(PULL_CONSUME_MSG_MQ_STATUS)) {
                String topic = uri.substring(index, uri.indexOf(PULL_CONSUME_MSG_MQ_STATUS));
                long timestamp = Long.parseLong(req.getParameter("timestamp"));
                Map<Integer, Long> stats = SpringContextHolder.getBean(KafkaPullConsumeManager.class).listPartitionStats(topic, timestamp);
                ResponseHelper.doWrite(resp, stats);
            } else {
                ResponseHelper.doError(resp, ErrorCode.noPermission.getCode(),
                        "http.method_post_not_supported from " + RequestHelper.getRemoteAddress(req));
            }
        } catch (Throwable t) {
            log.error("rocketmq doPost error: ", t);
            ResponseHelper.doError(resp, ErrorCode.systemInternalError.getCode(), t.getMessage());
        }
    }

    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        doPost(req, resp);
    }

}
